package com.alibaba.datax.plugin.writer.miniowriter.handler;

import com.alibaba.datax.common.element.Record;
import com.alibaba.datax.common.exception.DataXException;
import com.alibaba.datax.common.plugin.RecordReceiver;
import com.alibaba.datax.common.plugin.TaskPluginCollector;
import com.alibaba.datax.common.util.Configuration;
import com.alibaba.datax.plugin.unstructuredstorage.writer.Constant;
import com.alibaba.datax.plugin.unstructuredstorage.writer.TextCsvWriterManager;
import com.alibaba.datax.plugin.unstructuredstorage.writer.UnstructuredWriter;
import com.alibaba.datax.plugin.writer.miniowriter.Key;
import com.alibaba.datax.plugin.writer.miniowriter.MinioWriterErrorCode;

import java.io.IOException;
import java.io.OutputStream;
import java.io.Writer;
import java.util.List;

/**
 * @Author JDragon
 * @Date 2021.07.12 下午 2:23
 * @Email 1061917196@qq.com
 * @Des:
 */
public class CsvTxtWriterHandler extends WriterHandler {

    private Writer writer;

    private String fileFormat;

    private UnstructuredWriter unstructuredWriter;

    public CsvTxtWriterHandler(Writer writer, String fileFormat) {
        this.writer = writer;
        this.fileFormat = fileFormat;
    }

    @Override
    public void startWrite() {
        char fieldDelimiter = configuration.getChar(Key.FIELD_DELIMITER, Constant.DEFAULT_FIELD_DELIMITER);
        unstructuredWriter = TextCsvWriterManager.produceUnstructuredWriter(fileFormat, fieldDelimiter, writer);
        super.startWrite();
    }

    @Override
    public void writerHeader(List<String> header) {
        try {
            unstructuredWriter.writeOneRecord(header);
        } catch (IOException e) {
            throw DataXException.asDataXException(MinioWriterErrorCode.CSV_WRITE_TITLE_ERROR, e);
        }
    }

    @Override
    public void write(Record record) {
        try {
            List<String> dataList = resolveRow(record);
            unstructuredWriter.writeOneRecord(dataList);
        } catch (IOException e) {
            collectDirtyRecord(record, e);
        }
    }

    @Override
    public void finish() {

    }
}
